package ex.tableapi;

import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.Tumble;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.lit;

/**
 * https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/table/tableapi/#groupby-window-aggregation
 */
public class Ex04 extends ApiFrame{
    private String outName = "printOutTable";
    private    String createPrintOutDDL = "CREATE TABLE printOutTable (" +
            "a STRING, " +
            "b1  TIMESTAMP(3), " +
            "b2  TIMESTAMP(3), " +
            "b3  TIMESTAMP(3), " +
            "d  BIGINT " +
            ") WITH (" +
            " 'connector' = 'print' " +
            ")";

    public static void main(String[] args) {
        Ex04 demo =new Ex04();
        demo.getEnv();
        Table orders = demo.registerMysql("myorder", "orders");
        demo.registerConsole(demo.createPrintOutDDL,demo.outName);


        Table result = orders
                .window(Tumble.over(lit(5).minutes()).on($("rowtime")).as("w")) // define window
                .groupBy($("a"), $("w")) // group by key and window
                // access window properties and aggregate
                .select(
                        $("a"),
                        $("w").start().as("b1") ,
                        $("w").end().as("b2"),
                        $("w").rowtime().as("b3"),
                        $("b").sum().as("d")
                );

        result.printSchema();
        result.executeInsert(demo.outName);
    }
}
